package secsmarts;

import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.FilenameFilter;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

public class copyFile {

	public static void copy(String dir, String file) throws IOException {
		String[] args = { dir + "/" + file, dest };
		String path = parse(file);
		args[1] += path + "/" + file;
		if (args.length != 2) {
			System.err.println("Usage: filecopy <source> <target>");
			System.exit(2);
		}
		Configuration conf = new Configuration();
		conf.addResource("core-site.xml");
		conf.addResource("hdfs-site.xml");
		FileSystem fs;

		InputStream in = new BufferedInputStream(new FileInputStream(args[0]));
		fs = FileSystem.get(URI.create(args[1]), conf);
		OutputStream out = fs.create(new Path(args[1]));
		IOUtils.copyBytes(in, out, conf);

	}

	private static String parse(String f) {
		// TODO Auto-generated method stub
		// 输入："Bing_mUrls_2015_05_25_2.tsv"，输出：20150525
		return f.replace("Bing_mUrls_", "").substring(0, 11).replace("_", "");
		// return f.substring(12, 16) + f.substring(17, 18) + f.substring(19,
		// 20);

	}

	static Set<String> uploaded = new HashSet<String>();
	static String dest="hdfs://10.1.1.181:8020/data/bingmurls/";
	static String fileName = "uploaded";
	public static void main(String[] args) {
		String dir = args[0];
		dest=args[1];
		
		if (new File(fileName).exists())
			readFileByLines(fileName);
		File f = new File(dir);
		if (f.exists() && f.isDirectory()) {
			String[] fs = f.list(new FilenameFilter() {

				public boolean accept(File dir, String name) {
					// TODO Auto-generated method stub
					if (name.endsWith("tsv"))
						return true;
					return false;
				}
			});
			Arrays.sort(fs);
			upload(dir, fs);
		}
		saveUploaded(fileName);
	}

	private static void upload(String dir, String[] fs) {
		// TODO Auto-generated method stub
		for (String f : fs) {
			try {
				System.out.println(String.format("uploading file %s ", f));
				if (uploaded.contains(f))
					System.out.println(String.format("file %s uploaded,skip it", f));
				else {
					copy(dir, f);					
					uploaded.add(f);
					System.out.println(String.format("file %s upload,done", f));
					saveUploaded(fileName);
				}
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

		}
	}

	private static void saveUploaded(String fileName) {
		// TODO Auto-generated method stub
		File file = new File(fileName);
		BufferedWriter writer = null;
		try {
			
			writer = new BufferedWriter(new FileWriter(file));
			String tempString = null;
			int line = 1;
			// 一次读入一行，直到读入null为文件结束
			for (String str : uploaded)
				writer.write(str + "\n");
			writer.close();
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			if (writer != null) {
				try {
					writer.close();
				} catch (IOException e1) {
				}
			}
		}
	}

	public static void readFileByLines(String fileName) {
		File file = new File(fileName);
		BufferedReader reader = null;
		try {
			
			reader = new BufferedReader(new FileReader(file));
			String tempString = null;
			int line = 1;
			// 一次读入一行，直到读入null为文件结束
			while ((tempString = reader.readLine()) != null) {
				// 显示行号
				uploaded.add(tempString);
				System.out.println("line " + line + ": " + tempString);
				line++;
			}
			reader.close();
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			if (reader != null) {
				try {
					reader.close();
				} catch (IOException e1) {
				}
			}
		}
	}
}
